package org.example.listener;

import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.example.domain.MessageDO;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Argument;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class ConsumerListener {


	/**
	 * 配置监听的哪一个队列，同时在没有queue和exchange的情况下会去创建并建立绑定关系
	 * @param messageDO
	 * @param headers
	 * @param channel
	 * @throws IOException
	 */
	@RabbitListener(bindings = @QueueBinding(
			value = @Queue(value = "${cluster.name:demo}_" + "test_fanout_queue", durable = "true"),
			exchange = @Exchange(name = "${cluster.name:demo}_"+ "test_fanout_exchange"  , type = ExchangeTypes.FANOUT)
	)
	)
	@RabbitHandler
	public void test(@Payload MessageDO messageDO, @Headers Map<String,Object> headers, Channel channel) throws IOException {
		log.info("test msg:{}" ,messageDO.toString());
		Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);

		boolean multiple = false;

		//ACK,确认一条消息已经被消费 deliveryTag:该消息的index, multiple:是否批量.true:将一次性ack所有小于deliveryTag的消息。
		channel.basicAck(deliveryTag, multiple);
	}

	@RabbitListener(bindings = @QueueBinding(
			value = @Queue(value = "${cluster.name:demo}_" + "test_direct_queue", durable = "true"),
			exchange = @Exchange(name = "${cluster.name:demo}_"+ "test_direct_exchange" , type = ExchangeTypes.DIRECT)
	)
	)
	@RabbitHandler
	public void testDirect(@Payload MessageDO messageDO, @Headers Map<String,Object> headers, Channel channel) throws IOException {
		log.info("test direct msg:{}" ,messageDO.toString());
		Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);

		boolean multiple = false;

		//ACK,确认一条消息已经被消费 deliveryTag:该消息的index, multiple:是否批量.true:将一次性ack所有小于deliveryTag的消息。
		channel.basicAck(deliveryTag, multiple);
	}

	/**
	 * 延时队列
	 * @param messageDO
	 * @param headers
	 * @param channel
	 * @throws IOException
	 */
	@RabbitListener(bindings= @QueueBinding(value = @Queue(value = "${cluster.name:demo}_" + "test_delay_queue", durable = "true"),
			exchange = @Exchange(value = "${cluster.name:demo}_"+ "test_delay_exchange", type= ExchangeTypes.DIRECT,
					arguments=@Argument(name="x-delayed-type",value="direct"), delayed= Exchange.TRUE),
			key = "x-delayed-message"))
	@RabbitHandler
	public void testDelay(@Payload MessageDO messageDO, @Headers Map<String,Object> headers, Channel channel) throws IOException {
		log.info("test delay msg:{}" ,messageDO.toString());
		Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);

		boolean multiple = false;

		//ACK,确认一条消息已经被消费 deliveryTag:该消息的index, multiple:是否批量.true:将一次性ack所有小于deliveryTag的消息。
		channel.basicAck(deliveryTag, multiple);
	}

}
